import com.mysql.jdbc.Driver;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Test;

import java.io.IOException;


import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.List;

public class MysqlIndex {
    /**
     * 给hbase写数据
     */
    @Test
    public void write() throws Exception {
        //模拟数据
        for (int i = 11; i < 20; i++) {
            String rowkey = "100"+i;
            String name = "wzx---"+i;
            String age = (20+i)+"";
            //1.将数据保存到hbase
            writeToHBase(rowkey,name,age);
            //2.将经常要查询的字段及原rowkey写入mysql
            writeToMysql(rowkey,name);
        }

    }
    public void writeToHBase(String rowkey,String name,String age) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        //1.获取zk地址
        conf.set("hbase.zookeeper.quorum","hadoop102:2181,hadoop103:2181,hadoop104:2181");
        //2.获取hbase连接对象
        Connection connection = ConnectionFactory.createConnection(conf);
        //3.获取表对象
        Table table = connection.getTable(TableName.valueOf("STUDENT"));
        //4.封装数据
        Put put = new Put(rowkey.getBytes());
        put.addColumn("INFO1".getBytes(),"NAME".getBytes(),name.getBytes());
        put.addColumn("INFO1".getBytes(),"age".getBytes(),age.getBytes());
        //5.    插入数据
        table.put(put);
        //关闭
        table.close();
        connection.close();
    }
    public void writeToMysql(String rowkey,String name) throws Exception {
        //1.jdbc连接mysql的driver
       Class.forName("com.mysql.jdbc.Driver");
        //2.获取mysql的连接对象
        java.sql.Connection connection = DriverManager.getConnection
                ("jdbc:mysql://hadoop102:3306/practice", "root", "123321");
        //3.获取预编译对象
        PreparedStatement statement = connection.prepareStatement
                ("insert into student_index values(?,?)");
        //4.进行预编译
        statement.setString(1,rowkey);
        statement.setString(2,name);
        //5.执行操作
        statement.executeUpdate();
        //6.关闭资源
        statement.close();
        connection.close();
    }

    /**
     * 查找hbase的数据
     */
    @Test
    public void read() throws Exception {
        String name = "wzx---8";
        //1.去mysql查询索引得到原rowkey
        String rowkey = readByMysql(name);
        //2.通过原rowkey去HBase查找数据
        readByHBase(rowkey);
    }
    public void readByHBase(String rowkey) throws IOException {
        Configuration conf = HBaseConfiguration.create();
        //1.获取zk地址
        conf.set("hbase.zookeeper.quorum","hadoop102:2181,hadoop103:2181,hadoop104:2181");
        //2.获取hbase连接对象
        Connection connection = ConnectionFactory.createConnection(conf);
        //3.获取表对象
        Table table = connection.getTable(TableName.valueOf("STUDENT"));
        //4.封装rowkey
        Get get = new Get(rowkey.getBytes());
        //5.根据rowkey查询数据
        Result result = table.get(get);
        //6.遍历cell
        List<Cell> cells = result.listCells();
        for (Cell cell : cells) {
            String family = Bytes.toString(CellUtil.cloneFamily(cell));
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
            String value = Bytes.toString(CellUtil.cloneValue(cell));
            System.out.println("family:"+family+",qualifier:"+qualifier+",value:"+value);
        }
        //7.关闭资源
        table.close();
        connection.close();
    }

    public String readByMysql(String name) throws Exception {
        //1.jdbc连接mysql的driver
        Class.forName("com.mysql.jdbc.Driver");
        //2.获取mysql的连接对象
        java.sql.Connection connection = DriverManager.getConnection(
                "jdbc:mysql://hadoop102:3306/practice", "root", "123321");
        //3.获取预编译对象
        PreparedStatement statement = connection.prepareStatement(
                "select rowkey from student_index where name=?");
        //4.进行预编译
        statement.setString(1,name);
        //5.获取查询结果rowkey
        ResultSet resultSet = statement.executeQuery();
        String string = null;
        while(resultSet.next()){
             string = resultSet.getString("rowkey");
        }
        //5.关闭资源
        statement.close();
        connection.close();
        return string;
    }

}
